package com.capgemini.daos;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.capgemini.data.Athlete;
import com.capgemini.data.Discipline;
import com.capgemini.data.OlympicAdress;
import com.capgemini.data.Performance;
import com.capgemini.data.Person;
import com.capgemini.data.Tournament;
import com.capgemini.data.Trainer;

public class PersonDao extends Dao<Person> {

	private long perscounter;
	private long adcounter;

	public PersonDao() {
		perscounter = 0;
		adcounter = 0;
	}

	public void addRow(Person type) throws ParseException {
		try {
			HTable table = new HTable(conf, "Person");
			Person past = null;
			long persid;
			Put put;
			String pre = "";
			if (type.getClass().equals(Athlete.class)) {
				pre = "atl/";
			} else {
				pre = "tra/";
			}
			if (type.getId() == null || type.getId() == 0) {
				perscounter++;
				persid = perscounter;
			} else {
				persid = type.getId();
				past = this.getRecord(pre + String.valueOf(persid));
			}
			put = new Put(Bytes.toBytes(pre + String.valueOf(persid)));
			put.add(Bytes.toBytes("PersonInfo"), Bytes.toBytes("name"),
					Bytes.toBytes(type.getName()));
			put.add(Bytes.toBytes("PersonInfo"), Bytes.toBytes("surName"),
					Bytes.toBytes(type.getSurName()));
			put.add(Bytes.toBytes("PersonInfo"), Bytes.toBytes("birthDate"),
					Bytes.toBytes(type.getBirthDate().toString()));
			put.add(Bytes.toBytes("PersonInfo"), Bytes.toBytes("nationality"),
					Bytes.toBytes(type.getNationality()));
			if (type.getAdress() != null) {
				if (type.getAdress().getId() == null
						|| type.getAdress().getId() == 0) {
					adcounter++;
					type.getAdress().setId(adcounter);
				}
				put.add(Bytes.toBytes("OlympicAddress"),
						Bytes.toBytes("address"),
						Bytes.toBytes(type.getAdress().toPersistString()));
			} else {
				if (past != null) {
					if (past.getAdress() != null) {
						Delete deloa = new Delete(Bytes.toBytes(pre
								+ String.valueOf(persid)));
						deloa.deleteFamily(Bytes.toBytes("OlympicAddress"));
						table.delete(deloa);
					}
				}
			}

			if (type.getClass().equals(Athlete.class)) {
				put.add(Bytes.toBytes("AthleteInfo"), Bytes
						.toBytes("athleteNumber"), Bytes.toBytes(String
						.valueOf((((Athlete) type).getAthleteNumber()))));
				put.add(Bytes.toBytes("AthleteInfo"), Bytes.toBytes("sport"),
						Bytes.toBytes(((Athlete) type).getSport().toString()));

				long dopcounter = 0;
				for (String tempstring : ((Athlete) type).getDopingHistory()) {
					dopcounter++;
					put.add(Bytes.toBytes("AthleteInfo"),
							Bytes.toBytes("doping/"
									+ String.valueOf(dopcounter)),
							Bytes.toBytes(tempstring));
				}
				dopcounter++;
				if (past != null) {
					while (((Athlete) past).getDopingHistory().size() >= dopcounter) {
						Delete deldop = new Delete();
						deldop.deleteColumns(
								Bytes.toBytes("dopingHistory"),
								Bytes.toBytes("doping/"
										+ String.valueOf(dopcounter)));
						table.delete(deldop);
					}
				}
				Delete del = new Delete(
						(pre + String.valueOf(persid)).getBytes());
				del.deleteFamily(Bytes.toBytes("Performance"));
				table.delete(del);
				if (((Athlete) type).getPerformances() != null) {
					for (String templong : ((Athlete) type).getPerformances()) {
						put.add(Bytes.toBytes("Performance"),
								Bytes.toBytes(templong),
								Bytes.toBytes(templong));
					}
				}

				if (past != null) {
					for (Long temp : ((Athlete) past).getTrainers()) {
						if (((Athlete) type).getTrainers() == null) {
							put.add(Bytes.toBytes("Athlete"),
									Bytes.toBytes(String.valueOf(temp)), null);
						} else {
							if (!((Athlete) type).getTrainers().contains(temp)) {
								Put p = new Put(
										("tra/" + String.valueOf(temp))
												.getBytes());
								p.add(Bytes.toBytes("Athlete"),
										Bytes.toBytes(String.valueOf(persid)),
										null);
								table.put(p);
							}
						}
					}
				}
				Delete deltra = new Delete(
						(pre + String.valueOf(persid)).getBytes());
				deltra.deleteFamily(Bytes.toBytes("Trainer"));
				table.delete(deltra);
				for (long templong : ((Athlete) type).getTrainers()) {
					put.add(Bytes.toBytes("Trainer"),
							Bytes.toBytes(String.valueOf(templong)),
							Bytes.toBytes(String.valueOf(templong)));
					Put p = new Put(("tra/" + templong).getBytes());
					p.add(Bytes.toBytes("Athlete"),
							Bytes.toBytes(String.valueOf(persid)),
							Bytes.toBytes(String.valueOf(persid)));
					table.put(p);
				}

				// Trainer
			} else {
				put.add(Bytes.toBytes("TrainerInfo"), Bytes
						.toBytes("formerAthlete"), Bytes.toBytes(String
						.valueOf(((Trainer) type).isFormerAthlete())));
				put.add(Bytes.toBytes("TrainerInfo"),
						Bytes.toBytes("trainerHierarchy"),
						Bytes.toBytes(((Trainer) type).getTrainerHierarchy()));

				if (past != null) {
					for (Long temp : ((Trainer) past).getAthletes()) {
						if (!((Trainer) type).getAthletes().contains(temp)) {
							Put p = new Put(
									("atl/" + String.valueOf(temp)).getBytes());
							p.add(Bytes.toBytes("Trainer"),
									Bytes.toBytes(String.valueOf(persid)), null);
							table.put(p);
						}
					}
				}
				Delete delat = new Delete(
						(pre + String.valueOf(persid)).getBytes());
				delat.deleteFamily(Bytes.toBytes("Athlete"));
				table.delete(delat);
				for (long templong : ((Trainer) type).getAthletes()) {
					put.add(Bytes.toBytes("Athlete"),
							Bytes.toBytes(String.valueOf(templong)),
							Bytes.toBytes(templong));
					Put p = new Put(("atl/" + templong).getBytes());
					p.add(Bytes.toBytes("Trainer"),
							Bytes.toBytes(String.valueOf(persid)),
							Bytes.toBytes(String.valueOf(persid)));
					table.put(p);
				}
			}

			table.put(put);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static Person getRecord(String rowkey) throws IOException,
			ParseException {
		HTable table = new HTable(conf, "Person");
		Get get = new Get(rowkey.getBytes());
		Result rs = table.get(get);
		return readPerson(rs);
	}

	public static Person readPerson(Result r) throws ParseException {
		Person temp;
		String init = new String(r.getRow());
		init = init.substring(0, 3);
		if (init.equals("atl")) {
			temp = new Athlete();
			((Athlete) temp).setSport(Discipline.valueOf(new String(r.getValue(
					Bytes.toBytes("AthleteInfo"), Bytes.toBytes("sport")))));
			((Athlete) temp).setAthleteNumber(Integer.parseInt(new String(r
					.getValue(Bytes.toBytes("AthleteInfo"),
							Bytes.toBytes("athleteNumber")))));
			int dopcount = 1;
			while (r.containsColumn(Bytes.toBytes("AthleteInfo"),
					Bytes.toBytes("doping/" + dopcount))) {
				((Athlete) temp).addDopingHistory(new String(r.getValue(
						Bytes.toBytes("AthleteInfo"),
						Bytes.toBytes("doping/" + dopcount))));
				dopcount++;
			}
			NavigableMap<byte[], byte[]> perfmap = r.getFamilyMap(Bytes
					.toBytes("Performance"));
			for (byte[] tempbyte : perfmap.values()) {
				String tempstr = new String(tempbyte);
				((Athlete) temp).addPerformance(tempstr);
			}
			NavigableMap<byte[], byte[]> trainermap = r.getFamilyMap(Bytes
					.toBytes("Trainer"));
			for (byte[] train : trainermap.values()) {
				((Athlete) temp).addTrainer(Long.parseLong(new String(train)));
			}
		} else {
			temp = new Trainer();
			((Trainer) temp).setFormerAthlete(Boolean.getBoolean(new String(r
					.getValue(Bytes.toBytes("TrainerInfo"),
							Bytes.toBytes("formerAthlete")))));
			((Trainer) temp).setFormerAthlete(Boolean.getBoolean(new String(r
					.getValue(Bytes.toBytes("TrainerInfo"),
							Bytes.toBytes("formerAthlete")))));
			NavigableMap<byte[], byte[]> trainermap = r.getFamilyMap(Bytes
					.toBytes("Athlete"));
			for (byte[] tempat : trainermap.values()) {
				((Trainer) temp)
						.addAthletes(Long.parseLong(new String(tempat)));
			}
		}
		temp.setId(Long.valueOf((new String(r.getRow())).substring(4)));
		temp.setName(new String(r.getValue(Bytes.toBytes("PersonInfo"),
				Bytes.toBytes("name"))));
		temp.setSurName(new String(r.getValue(Bytes.toBytes("PersonInfo"),
				Bytes.toBytes("surName"))));
		temp.setNationality(new String(r.getValue(Bytes.toBytes("PersonInfo"),
				Bytes.toBytes("nationality"))));

		String temps = new String(r.getValue(Bytes.toBytes("PersonInfo"),
				Bytes.toBytes("birthDate")));

		DateFormat formater = new SimpleDateFormat("yyyy-MM-dd");
		java.util.Date parsedUtilDate = formater.parse(temps);
		temp.setBirthDate(new java.sql.Date(parsedUtilDate.getTime()));

		if (r.containsColumn(Bytes.toBytes("OlympicAddress"),
				Bytes.toBytes("address"))) {
			OlympicAdress newoa = new OlympicAdress();
			String str = new String(r.getValue(Bytes.toBytes("OlympicAddress"),
					Bytes.toBytes("address")));
			String tempstr = str.substring(18, str.indexOf(','));
			if (tempstr.equals("null")) {
				newoa.setId((long) 0);
			} else {
				newoa.setId(Long.parseLong(tempstr));
			}
			str = str.substring(str.indexOf(',') + 13);
			tempstr = str.substring(0, str.indexOf(','));
			newoa.setStreetName(tempstr);
			str = str.substring(str.indexOf(',') + 15);
			tempstr = str.substring(0, str.indexOf(','));
			newoa.setStreetNumber((Integer.parseInt(tempstr)));
			str = str.substring(str.indexOf(',') + 7);
			tempstr = str.substring(0, str.indexOf(']'));
			newoa.setCity((tempstr));
			newoa.setPerson(temp);
			temp.setAdress(newoa);
		}
		return temp;
	}

	public static void delRecord(String tableName, String rowKey)
			throws Exception {
		HTable table = new HTable(conf, tableName);
		Person temp = getRecord(rowKey);
		TournamentDao tourdao = new TournamentDao();
		if (rowKey.substring(0, 3).equals("atl")) {
			String tourstring;
			Tournament temptour;
			String perfid;
			for (String perf : ((Athlete) temp).getPerformances()) {
				tourstring = perf.substring(0, perf.indexOf('/'));
				perfid = perf.substring(perf.indexOf('/') + 1);
				temptour = tourdao.getRecord(tourstring);
				for (Performance perfo : temptour.getPerformances()) {
					if (perfo.getId() == Long.valueOf(perfid)) {
						temptour.getPerformances().remove(perfo);
						tourdao.addRow(temptour);
					}
				}
			}
			for (Long trainer : ((Athlete) temp).getTrainers()) {
				Delete del = new Delete(("tra/" + trainer).getBytes());
				del.deleteColumns("Athlete".getBytes(), rowKey.substring(4)
						.getBytes());
				table.delete(del);
			}
		} else {
			for (Long athlete : ((Trainer) temp).getAthletes()) {
				Delete del = new Delete(("atl/" + athlete).getBytes());
				del.deleteColumns("Trainer".getBytes(), rowKey.substring(4)
						.getBytes());
				table.delete(del);
			}
		}

		Delete del = new Delete(rowKey.getBytes());
		table.delete(del);
	}
}
